public static Optional<MultiInputShape> detect(RelNode node) {
     List<RelNode> inputs = node.getInputs();
     if (inputs.size() < 2) {
         return Optional.empty();
     }
     List<OpenSearchStageInputScan> scans = new ArrayList<>(inputs.size());
     for (RelNode input : inputs) {
-        if (!(input instanceof OpenSearchExchangeReducer reducer)) {
+        RelNode unwrappedInput = input instanceof org.apache.calcite.plan.hep.HepRelVertex v ? v.getCurrentRel() : input;
+        if (!(unwrappedInput instanceof OpenSearchExchangeReducer reducer)) {
             return Optional.empty();
         }
-        if (!(reducer.getInput() instanceof OpenSearchStageInputScan scan)) {
+        RelNode reducerInput = reducer.getInput() instanceof org.apache.calcite.plan.hep.HepRelVertex v2 ? v2.getCurrentRel() : reducer.getInput();
+        if (!(reducerInput instanceof OpenSearchStageInputScan scan)) {
             return Optional.empty();
         }
         scans.add(scan);
     }
     return Optional.of(new MultiInputShape(scans));
 }

 TargetResolver targetResolver = grandchildren.isEmpty() ? new ShardTargetResolver(childFragment, clusterService) : null;
 ExchangeSinkProvider childSinkProvider = null;
 if (!grandchildren.isEmpty()) {
     List<String> reduceViable = CapabilityResolutionUtils.filterByReduceCapability(registry, reducer.getViableBackends());
+    if (reduceViable.isEmpty()) {
+        throw new IllegalStateException(
+            "No backend supports reduce capability among viable backends: " + reducer.getViableBackends()
+        );
+    }
     childSinkProvider = registry.getBackend(reduceViable.getFirst()).getExchangeSinkProvider();
 }

 if (viableBackends.isEmpty()) {
-    throw new IllegalStateException("No backend supports JOIN among viable backends after intersecting inputs");
+    return; // no backend supports JOIN — let the planner report a proper failure
 }

-private static boolean dataProvisioned = false;
+private static volatile boolean dataProvisioned = false;
 
-/**
- * Lazily provision both calcs indices on first invocation. Called inside test
- * methods — {@code client()} is not available in {@code @BeforeClass}.
- */
-private void ensureDataProvisioned() throws IOException {
+private synchronized void ensureDataProvisioned() throws IOException {
     if (dataProvisioned == false) {
         DatasetProvisioner.provision(client(), CALCS);
         DatasetProvisioner.provision(client(), CALCS_ALT);
         dataProvisioned = true;
     }
 }

 public static Optional<MultiInputShape> detect(RelNode node) {
     List<RelNode> inputs = node.getInputs();
     if (inputs.size() < 2) {
         return Optional.empty();
     }
     List<OpenSearchStageInputScan> scans = new ArrayList<>(inputs.size());
     for (RelNode input : inputs) {
-        if (!(input instanceof OpenSearchExchangeReducer reducer)) {
+        RelNode unwrapped = input instanceof HepRelVertex v ? v.getCurrentRel() : input;
+        if (!(unwrapped instanceof OpenSearchExchangeReducer reducer)) {
             return Optional.empty();
         }
-        if (!(reducer.getInput() instanceof OpenSearchStageInputScan scan)) {
+        RelNode reducerInput = reducer.getInput() instanceof HepRelVertex v ? v.getCurrentRel() : reducer.getInput();
+        if (!(reducerInput instanceof OpenSearchStageInputScan scan)) {
             return Optional.empty();
         }
         scans.add(scan);
     }
     return Optional.of(new MultiInputShape(scans));
 }

 TargetResolver targetResolver = grandchildren.isEmpty() ? new ShardTargetResolver(childFragment, clusterService) : null;
 ExchangeSinkProvider childSinkProvider = null;
 if (!grandchildren.isEmpty()) {
     List<String> reduceViable = CapabilityResolutionUtils.filterByReduceCapability(registry, reducer.getViableBackends());
+    if (reduceViable.isEmpty()) {
+        throw new IllegalStateException("No backend supports reduce capability among viable backends: " + reducer.getViableBackends());
+    }
     childSinkProvider = registry.getBackend(reduceViable.getFirst()).getExchangeSinkProvider();
 }

 List<String> viableBackends = computeViableBackends(join.getLeft(), join.getRight());
+if (viableBackends.isEmpty()) {
+    return; // inputs not yet marked — let HEP retry after child rules fire
+}
 List<String> joinCapable = context.getCapabilityRegistry().operatorBackends(EngineCapability.JOIN);
 viableBackends.retainAll(joinCapable);
 if (viableBackends.isEmpty()) {
     throw new IllegalStateException("No backend supports JOIN among viable backends after intersecting inputs");
 }

-private static boolean dataProvisioned = false;
+private static volatile boolean dataProvisioned = false;
 
-private void ensureDataProvisioned() throws IOException {
+private synchronized void ensureDataProvisioned() throws IOException {
     if (dataProvisioned == false) {
         DatasetProvisioner.provision(client(), CALCS);
         DatasetProvisioner.provision(client(), CALCS_ALT);
         dataProvisioned = true;
     }
 }

 TargetResolver targetResolver = grandchildren.isEmpty() ? new ShardTargetResolver(childFragment, clusterService) : null;
 ExchangeSinkProvider childSinkProvider = null;
 if (!grandchildren.isEmpty()) {
     List<String> reduceViable = CapabilityResolutionUtils.filterByReduceCapability(registry, reducer.getViableBackends());
+    if (reduceViable.isEmpty()) {
+        throw new IllegalStateException(
+            "No backend supports reduce capability for stage "
+                + childStageId
+                + "; viable backends were: "
+                + reducer.getViableBackends()
+        );
+    }
     childSinkProvider = registry.getBackend(reduceViable.getFirst()).getExchangeSinkProvider();
 }

-private volatile boolean childClosed;
+private final java.util.concurrent.atomic.AtomicBoolean childClosed = new java.util.concurrent.atomic.AtomicBoolean(false);

-} catch (IllegalStateException closed) {
-    // QueryContext closed while this response was in flight — another
-    // shard failed and tore down teardown. Silently drop; onShardTerminated
-    // still needs to run so inFlight drains and the stage's terminal
-    // transition isn't wedged.
+} catch (Exception ex) {
+    // Capture unexpected errors so the stage transitions to FAILED.
+    // For the "sink already closed" case (QueryContext torn down) we
+    // still need onShardTerminated() to drain inFlight.
+    captureFailure(ex);
     if (isLast) {
         onShardTerminated();
     }
 }

 @Override
 public void close() {
-    if (childClosed) {
+    if (!childClosed.compareAndSet(false, true)) {
         return;
     }
-    childClosed = true;
     try {
         sender.close();
     } catch (Throwable t) {
         logger.warn("[ReduceSink] error closing child sender", t);
     }
 }

 public static Optional<MultiInputShape> detect(RelNode node) {
     List<RelNode> inputs = node.getInputs();
     if (inputs.size() < 2) {
         return Optional.empty();
     }
     List<OpenSearchStageInputScan> scans = new ArrayList<>(inputs.size());
     for (RelNode input : inputs) {
         if (!(input instanceof OpenSearchExchangeReducer reducer)) {
             return Optional.empty();
         }
-        if (!(reducer.getInput() instanceof OpenSearchStageInputScan scan)) {
-            return Optional.empty();
+        RelNode reducerInput = reducer.getInput();
+        if (!(reducerInput instanceof OpenSearchStageInputScan scan)) {
+            // Reducer's input was not severed to a StageInputScan — this indicates
+            // an incomplete DAG cut. Fail fast rather than silently falling through
+            // to the wrong conversion path.
+            throw new IllegalStateException(
+                "Expected OpenSearchStageInputScan under ExchangeReducer in multi-input shape, got: "
+                    + reducerInput.getClass().getSimpleName()
+            );
         }
         scans.add(scan);
     }
     return Optional.of(new MultiInputShape(scans));
 }

 public OpenSearchUnion(RelOptCluster cluster, RelTraitSet traitSet, List<RelNode> inputs, boolean all, List<String> viableBackends) {
     super(cluster, traitSet, List.of(), inputs, all);
-    this.viableBackends = viableBackends;
+    this.viableBackends = List.copyOf(viableBackends);
 }

 new OpenSearchFilterRule(context),
 new OpenSearchProjectRule(context),
 new OpenSearchAggregateRule(context),
-new OpenSearchJoinRule(context),
 new OpenSearchSortRule(context),
-new OpenSearchUnionRule(context)
+new OpenSearchUnionRule(context),
+new OpenSearchJoinRule(context)

-private static boolean dataProvisioned = false;
+private static volatile boolean dataProvisioned = false;
 
-private void ensureDataProvisioned() throws IOException {
+private synchronized void ensureDataProvisioned() throws IOException {
     if (dataProvisioned == false) {
         DatasetProvisioner.provision(client(), CALCS);
         DatasetProvisioner.provision(client(), CALCS_ALT);
         dataProvisioned = true;
     }
 }

 private final class ChildSink implements ExchangeSink {
     private final DatafusionPartitionSender sender;
-    private volatile boolean childClosed;
+    private final java.util.concurrent.atomic.AtomicBoolean childClosed = new java.util.concurrent.atomic.AtomicBoolean(false);
     ...
     @Override
     public void close() {
-        if (childClosed) {
+        if (!childClosed.compareAndSet(false, true)) {
             return;
         }
-        childClosed = true;

-if (agreedBackend == null) {
+if (agreedBackend == null && childAlt.chosenBackend != null && !childAlt.chosenBackend.isEmpty()) {
     agreedBackend = childAlt.chosenBackend;
-} else if (childAlt.chosenBackend != null
+} else if (agreedBackend != null
+    && childAlt.chosenBackend != null
     && !childAlt.chosenBackend.isEmpty()
     && !childAlt.chosenBackend.equals(agreedBackend)) {

 // Operator above the final-fragment boundary — convert child first, then attach.
-byte[] innerBytes = convertReduceNode(node.getInputs().getFirst(), convertor, false, delegationBytes);
+List<RelNode> inputs = node.getInputs();
+if (inputs.size() != 1) {
+    throw new IllegalStateException(
+        "Expected exactly one input for attach-on-top node " + node.getClass().getSimpleName() + ", got " + inputs.size()
+    );
+}
+byte[] innerBytes = convertReduceNode(inputs.getFirst(), convertor, false, delegationBytes);
 return convertor.attachFragmentOnTop(strippedNode, innerBytes);

-// Register a reader even when no files were flushed (empty-shard case). Without
-// this, queries against an empty shard throw "No DataFusion reader available"
-// because refresh listeners are only invoked when didRefresh==true, but getReader
-// is called on every query regardless of whether any docs were ever indexed.
+if (didRefresh == false && readers.isEmpty() == false) return;
 if (readers.containsKey(catalogSnapshot)) return;

 } catch (IllegalStateException closed) {
     // QueryContext closed while this response was in flight — another
     // shard failed and tore down teardown. Silently drop; onShardTerminated
     // still needs to run so inFlight drains and the stage's terminal
     // transition isn't wedged.
+    logger.debug("Dropping in-flight response after sink closed", closed);
     if (isLast) {
         onShardTerminated();
     }
 }

 if (markedInputs.isEmpty()) {
     // Defensive — Calcite shouldn't construct a Union with all-empty inputs, but
     // surfacing a clear message beats letting downstream rules fail mysteriously.
-    throw new IllegalStateException("Union rule encountered Union with all-empty inputs");
+    throw new IllegalStateException(
+        "Union rule encountered Union with all-empty inputs (all " + union.getInputs().size()
+            + " inputs were empty Values nodes and were dropped)"
+    );
 }

-int singleChildStageId = childInputs.keySet().iterator().next();
 NativeBridge.registerMemtable(session.getPointer(), inputIdFor(singleChildStageId), schemaIpc, arrayPtrs, schemaPtrs);